package com.example.spider.config;

import org.shoulder.core.concurrent.Threads;
import org.shoulder.monitor.concurrent.MonitorableThreadPool;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @author lym
 */
@EnableAsync
@Configuration
public class ThreadPoolConfiguration {

    public static final String SPIDER = "spiderThreadPool";

    public static final String HANDLER = "handlerThreadPool";
    public static final String IP_PROXY_TESTER = "ipTesterThreadPool";

    public static final int taskQueueMaxSize = 100_000;

    @Bean(SPIDER)
    public MonitorableThreadPool spiderThreadPool() {
        return new MonitorableThreadPool(30, 50, 0L, TimeUnit.SECONDS,
                new LinkedBlockingDeque<>(taskQueueMaxSize), new MyThreadFactory("SPIDER", true),
                new ThreadPoolExecutor.CallerRunsPolicy(), SPIDER);
    }

    @Bean(HANDLER)
    public MonitorableThreadPool handlerThreadPool() {
        return new MonitorableThreadPool(30, 200, 60L, TimeUnit.SECONDS,
                new LinkedBlockingDeque<>(1000), new MyThreadFactory("HANDLER", true),
                new ThreadPoolExecutor.CallerRunsPolicy(), HANDLER);
    }

    /**
     * 校验由定时任务触发，拒绝策略不能使用 callerRun，可暂时丢弃
     */
    @Qualifier(ThreadPoolConfiguration.IP_PROXY_TESTER)
    @Bean(IP_PROXY_TESTER)
    public MonitorableThreadPool ipProxyTesterThreadPool() {
        return new MonitorableThreadPool(300, 1000, 30L, TimeUnit.SECONDS,
                new LinkedBlockingDeque<>(20000), new ThreadPoolConfiguration.MyThreadFactory("IpProxyTester", true),
                new ThreadPoolExecutor.DiscardPolicy(), IP_PROXY_TESTER);
    }

    @Bean(Threads.SHOULDER_THREAD_POOL_NAME)
    public ExecutorService shoulderThreadPool() {
        // 替代内置threadPool queue扩大，max扩大
        ThreadPoolExecutor executor = new MonitorableThreadPool(50, 50,
                60L, TimeUnit.SECONDS, new LinkedBlockingQueue<>(taskQueueMaxSize),
                new CustomizableThreadFactory("shoulder"), Threads.SHOULDER_THREAD_POOL_NAME);
        // 提前设置，方便使用
        Threads.setExecutorService(executor);
        return executor;
    }


    public static class MyThreadFactory implements ThreadFactory {

        private final ThreadGroup threadGroup;
        private final AtomicLong threadSeq = new AtomicLong(1);
        private final String threadNamePrefix;
        private boolean daemon = true;

        public MyThreadFactory(String groupName, boolean isDaemon) {
            this.daemon = isDaemon;
            this.threadNamePrefix = groupName;
            this.threadGroup = new ThreadGroup(groupName);
        }


        @Override
        public Thread newThread(Runnable r) {
            Thread t = new Thread(threadGroup, r, threadNamePrefix + threadSeq.getAndIncrement(), 0);
            t.setDaemon(daemon);
            //recovery thread priority
            if (t.getPriority() != Thread.NORM_PRIORITY) {
                t.setPriority(Thread.NORM_PRIORITY);
            }
            return t;
        }

    }

    //激活 @Timed 注解

    //@Bean
    //public TimedAspect timedAspect(MeterRegistry registry) {
    //    return new TimedAspect(registry);
    //}

}
